package kvraft

import (
	"sync"
	"sync/atomic"

	"../labgob"
	"../labrpc"
	"../raft"
)

type KVServer struct {
	mu      sync.Mutex
	me      int
	rf      *raft.Raft
	applyCh chan raft.ApplyMsg
	dead    int32 // set by Kill()

	maxraftstate int // snapshot if log grows this big

	// Your definitions here.
	clientsCount
	answerPlugin
	db KVDB
}

func (kv *KVServer) State(args *StateArgs, reply *StateReply) {
	if kv.killed() {
		reply.Err = ErrKilled
		return
	}

	term, isLeader := kv.rf.GetState()
	reply.Term = term
	reply.IsLeader = isLeader
	reply.Index = kv.me
}

func (kv *KVServer) Get(args *GetArgs, reply *GetReply) {
	if kv.killed() {
		reply.Err = ErrKilled
		return
	}

	op := GetArgs2Op(args)
	if _, _, isLeader := kv.rf.Start(*op); !isLeader {
		reply.Err = ErrWrongLeader
		DPrintf("[%d] Get: wrong leader", kv.me)
		return
	}
	gr := kv.WaitRaft(op)
	DPrintf("[%d] Get Op: %d end wait, return. err: %s, key: %s", kv.me, op.RequestId, gr.Err, gr.Value)
	if gr.Err == ErrNoKey {
		reply.Err = OK
		reply.Value = ""
		kv.RequestDone()
		return
	}
	reply.Err = gr.Err
	reply.Value = gr.Value
	kv.RequestDone()
}

func (kv *KVServer) PutAppend(args *PutAppendArgs, reply *PutAppendReply) {
	if kv.killed() {
		reply.Err = ErrKilled
		return
	}

	op := PutAppendArgs2Op(args)
	if _, _, isLeader := kv.rf.Start(*op); !isLeader {
		reply.Err = ErrWrongLeader
		return
	}
	gr := kv.WaitRaft(op)
	DPrintf("[%d] PutAppend Op: %d end wait, return. err: %s, value: %s", kv.me, op.RequestId, gr.Err, gr.Value)
	reply.Err = gr.Err
	reply.Value = gr.Value
	kv.RequestDone()
}

//func (kv *KVServer)

func (kv *KVServer) alwaysListenFromRaft() {
	for common := range kv.applyCh {
		DPrintf("[%d] get common", kv.me)
		op := common.Command.(Op)
		DPrintf("[%d] Op: %d, op.name: %s", kv.me, op.RequestId, op.Name)
		if e := kv.addIfOk(&op); e != OK {
			DPrintf("[%d] Op: %d has commit, pass it", kv.me, op.RequestId)
			kv.RaftDoneWaitRequest(&op, &generalReply{e, ""})
			continue
		}
		var gr *generalReply
		switch op.Name {
		case "get":
			gr = kv.db.Get(op.Key)
		case "put":
			gr = kv.db.Put(op.Key, op.Value)
		case "append":
			gr = kv.db.Append(op.Key, op.Value)
		default:
			gr = &generalReply{ErrNoOp, ""}
		}
		kv.RaftDoneWaitRequest(&op, gr)
	}
}

//
// the tester calls Kill() when a KVServer instance won't
// be needed again. for your convenience, we supply
// code to set rf.dead (without needing a lock),
// and a killed() method to test rf.dead in
// long-running loops. you can also add your own
// code to Kill(). you're not required to do anything
// about this, but it may be convenient (for example)
// Figure 8 (unreliable)to suppress debug output from a Kill()ed instance.
//
func (kv *KVServer) Kill() {
	atomic.StoreInt32(&kv.dead, 1)
	kv.rf.Kill()
	// Your code here, if desired.
}

func (kv *KVServer) killed() bool {
	z := atomic.LoadInt32(&kv.dead)
	return z == 1
}

//
// servers[] contains the ports of the set of
// servers that will cooperate via Raft to
// form the fault-tolerant key/value service.
// me is the index of the current server in servers[].
// the k/v server should store snapshots through the underlying Raft
// implementation, which should call persister.SaveStateAndSnapshot() to
// atomically save the Raft state along with the snapshot.
// the k/v server should snapshot when Raft's saved state exceeds maxraftstate bytes,
// in order to allow Raft to garbage-collect its log. if maxraftstate is -1,
// you don't need to snapshot.
// StartKVServer() must return quickly, so it should start goroutines
// for any long-running work.
//
func StartKVServer(servers []*labrpc.ClientEnd, me int, persister *raft.Persister, maxraftstate int) *KVServer {
	// call labgob.Register on structures you want
	// Go's RPC library to marshall/unmarshall.
	labgob.Register(Op{})

	kv := new(KVServer)
	kv.me = me
	kv.maxraftstate = maxraftstate

	// You may need initialization code here.
	kv.applyCh = make(chan raft.ApplyMsg)
	kv.rf = raft.Make(servers, me, persister, kv.applyCh)
	kv.clientsCount.init()
	kv.answerPlugin.Init()
	kv.db.Init(kv.me)
	go kv.alwaysListenFromRaft()

	return kv
}
